/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package edu.yonsei.iwb.ts.index;

//import au.com.bytecode.opencsv.CSVParser;
//import au.com.bytecode.opencsv.CSVReader;
import com.hp.hpl.jena.rdf.model.Literal;
import com.hp.hpl.jena.rdf.model.RDFNode;
import com.hp.hpl.jena.rdf.model.Resource;
import com.hp.hpl.jena.rdf.model.Statement;
import com.hp.hpl.jena.rdf.model.StmtIterator;
import com.hp.hpl.jena.vocabulary.OWL;
import com.hp.hpl.jena.vocabulary.RDF;
import com.hp.hpl.jena.vocabulary.RDFS;
import edu.yonsei.iwb.di.ESService;
import edu.yonsei.iwb.di.ESServiceConfig;
import edu.yonsei.iwb.ts.model.JenaResourceType;
import java.io.*;
import java.util.LinkedHashMap;

import org.apache.log4j.Logger;
import org.elasticsearch.ElasticsearchException;
import org.elasticsearch.action.WriteConsistencyLevel;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
import org.elasticsearch.action.bulk.BulkRequestBuilder;
import org.elasticsearch.action.index.IndexRequestBuilder;
import org.elasticsearch.client.transport.TransportClient;

import org.elasticsearch.common.settings.ImmutableSettings;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentBuilder;
import org.elasticsearch.common.xcontent.XContentFactory;

public class TSIndexer extends Thread {

    private TransportClient client;
    private File unindexedLogFile;
    private PrintWriter pw;
    private String indexname;
    private String filename;
    private StmtIterator stit;
    private boolean flag = true;
    private TSIndexerHandler ith;
    private LinkedHashMap<String, String> schemaMap;
    private Logger logger = Logger.getLogger(TSIndexer.class);

    public TSIndexer(TransportClient client, TSIndexerHandler ith, String indexname, String filename, StmtIterator stit) throws NullPointerException {
        this.ith = ith;
        this.client = client;
        this.filename = filename;
        this.indexname = indexname;
        this.stit = stit;
        this.client.admin().cluster().health(new ClusterHealthRequest("_all").waitForYellowStatus().timeout("3s")).actionGet();
    }

    @Override
    public void run() {
        try {
            logger.info("# start indexing [" + filename + "] ");            
            long startTime = System.currentTimeMillis();
            makeIndex();
            long endTime = System.currentTimeMillis();
            double indexingTime = (endTime - startTime) / 1000.0f;

            if (flag) {
                logger.info("# (" + Thread.currentThread().getName() + ") " + filename + " Indexing complete, Time = " + indexingTime + " sec");

            } else {
                logger.info("# (" + Thread.currentThread().getName() + ") " + filename + " Indexing stopped,  Time = " + indexingTime + " sec");
            }
        } catch (IOException ex) {
            logger.error("# (" + ex.getLocalizedMessage() + ") occured in indexing -" + filename);
        }
    }

    private void createIndex(String indexName) {
        try {
            Settings settings = ImmutableSettings.settingsBuilder()
                    .put("index.refresh_interval", -1)
                    .put("index.store.compress.stored", true)
                    .put("index.store.compress.tv", true)
                    .put("index.merge.policy.max_merge_at_once", 5)
                    .put("index.merge.policy.max_merged_segment", "1gb")
                    .put("index.merge.policy.segments_per_tier", 200)
                    .put("index.translog.flush_threshold_ops", 50000)
                    .put("index.translog.flush_threshold_size", "500mb")
                    .put("index.store.throttle.type", "merge")
                    .put("index.store.throttle.max_bytes_per_sec", "10mb")
                    .build();
            this.client.admin().indices().create(new CreateIndexRequest(indexName).settings(settings)).actionGet();
//            this.client.admin().indices().prepareUpdateSettings(indexName).setSettings(settings).execute().actionGet();
            this.client.admin().indices().preparePutMapping(indexName).setType(indexname).setSource(mappingBuilder()).execute().actionGet();
        } catch (ElasticsearchException ese) {
            logger.error(ese.getDetailedMessage());
        } catch (IOException ex) {
            logger.error(ex.getMessage());
        }

    }

    private void makeIndex() throws IOException {

        if (!this.client.admin().indices().prepareExists(indexname).execute().actionGet().isExists()) {
            this.createIndex(indexname);
        }

        String tempFileName = ESService.CONF.getProperty("data.directory.unindexed") + filename + "_unindexed";
        this.unindexedLogFile = new File(tempFileName);

        try {
            this.unindexedLogFile.createNewFile();
            this.pw = new PrintWriter(new BufferedOutputStream(new FileOutputStream(unindexedLogFile)));
        } catch (IOException io) {
            logger.error(tempFileName+ " / "+io.getMessage());
        }
        
        BulkRequestBuilder bulkRequest = client.prepareBulk();        
        stmtIterator:
        while (this.stit.hasNext()) {
            Statement st = this.stit.nextStatement();
//            System.out.println(st.toString());
            if(isUnused(st)){
                continue stmtIterator;
            }
//            System.out.println(st.toString());

            try {
                IndexRequestBuilder irb = buildDocument(st);
                bulkRequest.add(irb);
            } catch (IOException | NumberFormatException | ArrayIndexOutOfBoundsException e) {
                logger.error("# Failed to index a triple - " + st.getString() + "\n" + e.getMessage());
                this.pw.write(st.getSubject() + "," + st.getPredicate() + "," + st.getObject() + "\n");
            }

            if (bulkRequest.numberOfActions() == ESServiceConfig.BULK_ACTION_SIZE) {
                bulkRequest.execute().actionGet();
                bulkRequest = client.prepareBulk();
            }
        }

        if (bulkRequest.numberOfActions() != 0) {
            bulkRequest.execute().actionGet();
        }
        
        this.client.admin().indices().refresh(new RefreshRequest("_all")).actionGet();

        this.pw.flush();
        this.pw.close();
        if (this.unindexedLogFile.length() == 0) {
            this.unindexedLogFile.delete();
        }        
    }
    
    private boolean isUnused(Statement st){
        Resource subject = st.getSubject();
        if(UnIndexedResourceList.contain(subject)){
           return true;
        }else{
            return false;
        }
    }

    private IndexRequestBuilder buildDocument(Statement st) throws IOException, NumberFormatException, ArrayIndexOutOfBoundsException {
        IndexRequestBuilder irb = null;
        XContentBuilder docBuilder = XContentFactory.jsonBuilder().startObject();

        Resource subject = st.getSubject();
//        System.out.println("s : " + subject.toString());
        this.addFieldValue(docBuilder, TSIndexFields.SUBJECT, subject.toString());

        Resource predicate = st.getPredicate();
//        System.out.println("p : " + predicate.getURI());
        this.addFieldValue(docBuilder, TSIndexFields.PREDICATE, predicate.toString());
        
        RDFNode object = st.getObject();
        if(object.isResource()){
//            System.out.println("o : " + object.asResource().getURI());
//            System.out.println("o_type : resource");
            this.addFieldValue(docBuilder, TSIndexFields.OBJECT, object.asResource().toString());
            this.addFieldValue(docBuilder, TSIndexFields.OBJECT_TYPE, RDFS.Resource.toString());
        }else if(object.isLiteral()){
            Literal lit = object.asLiteral();
//            System.out.println("o : " + object.asLiteral().getLexicalForm());
//            System.out.println("o_type : " + object.asLiteral().getDatatype().toString());            
            
            this.addFieldValue(docBuilder, TSIndexFields.OBJECT, object.asLiteral().getLexicalForm());
            this.addFieldValue(docBuilder, TSIndexFields.OBJECT_TYPE, object.asLiteral().getDatatypeURI());
        }
            
        


        docBuilder.endObject();

        irb = this.client.prepareIndex(indexname, indexname).setConsistencyLevel(WriteConsistencyLevel.DEFAULT).setSource(docBuilder);
        return irb;
    }

    /**
     * Add field value into document
     *
     * @param docBuilder document builder
     * @param value field value
     * @param column field column id
     */
    private void addFieldValue(XContentBuilder docBuilder, String field, String value) {
        try {
            docBuilder.field(field, value);
        } catch (NumberFormatException e) {
            logger.error("# Value and Type is not correctly matched\n" + e.getMessage());
            throw e;
        } catch (ArrayIndexOutOfBoundsException e) {
            logger.error("# Value and Mapping Schema is not matched\n" + e.getMessage());
            throw e;
        } catch (IOException ex) {
            logger.error("# JSON Build error\n" + ex.getMessage());
        }
    }

//    private IndexRequestBuilder buildDocument(String s, String id, String indexType, String indexName) throws IOException {
//        IndexRequestBuilder irb = null;
//
//        XContentBuilder docBuilder = null;
//        docBuilder = XContentFactory.jsonBuilder().startObject();
//        addFieldValue(docBuilder, s, 0);
//        docBuilder.endObject();
//
//        irb = this.client.prepareIndex(indexName, indexType, id).setConsistencyLevel(WriteConsistencyLevel.DEFAULT).setSource(docBuilder);
//        return irb;
//    }
    private XContentBuilder mappingBuilder() throws IOException {

        XContentBuilder docBuilder = XContentFactory.jsonBuilder().prettyPrint().startObject().startObject(this.indexname);

        docBuilder.startObject("_source");
        docBuilder.field("compress", true);
        docBuilder.endObject();
        docBuilder.startObject("_all");
        docBuilder.field("enabled", false);
        docBuilder.endObject();

        docBuilder.startObject("properties");

        docBuilder.startObject(TSIndexFields.SUBJECT);
        docBuilder.field("type", "string");
        docBuilder.field("index", "not_analyzed");
        docBuilder.endObject();

        docBuilder.startObject(TSIndexFields.PREDICATE);
        docBuilder.field("type", "string");
        docBuilder.field("index", "not_analyzed");
        docBuilder.endObject();

        docBuilder.startObject(TSIndexFields.OBJECT);
        docBuilder.field("type", "string");
        docBuilder.field("index", "not_analyzed");
        docBuilder.endObject();

        docBuilder.startObject(TSIndexFields.OBJECT_TYPE);
        docBuilder.field("type", "string");
        docBuilder.field("index", "not_analyzed");
        docBuilder.endObject();

        docBuilder.endObject().endObject().endObject();
        return docBuilder;
    }

    public void stopIndexing() {
        this.flag = false;
    }
}

